package com.ctshk.common.elasticsearch.core;

import java.io.IOException;
import java.util.List;
import java.util.Map;

import com.alibaba.fastjson.JSON;
import com.ctshk.common.elasticsearch.constant.ElasticSearchConstant;
import com.ctshk.common.elasticsearch.entity.CusDemoInfoDocument;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.support.IndicesOptions;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.HttpAsyncResponseConsumerFactory;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

/**
 * 阿里云ES工具类
 */
@Component
public class ElasticsearchUtil {
    private static final Logger LOGGER = LoggerFactory.getLogger(ElasticsearchUtil.class);
    private static final RequestOptions COMMON_OPTIONS;
    @Value("${spring.elasticsearch.cluster_host}")
    private String clusterHost;
    @Value("${spring.elasticsearch.cluster_port}")
    private Integer clusterPort;
    @Value("${spring.elasticsearch.username}")
    private String username;
    @Value("${spring.elasticsearch.password}")
    private String password;

    static {
        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
        // 默认缓存限制为100MB，此处修改为30MB。
        builder.setHttpAsyncResponseConsumerFactory(
                new HttpAsyncResponseConsumerFactory
                        .HeapBufferedResponseConsumerFactory(30 * 1024 * 1024));
        COMMON_OPTIONS = builder.build();
    }

    /**
     * @desc 获取ES client
     */
    public RestHighLevelClient getEsClient() {
        Map<String, RestHighLevelClient> esClientMap = ElasticSearchConstant.esClientMap;
        if (esClientMap == null || esClientMap.isEmpty()) {
            // 阿里云Elasticsearch集群需要basic auth验证。
            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            // 访问用户名和密码,创建阿里云Elasticsearch实例时设置的用户名和密码，也是Kibana控制台的登录用户名和密码。
            credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(this.username, this.password));
            // 通过builder创建rest client，配置http client的HttpClientConfigCallback。
            // 单击所创建的Elasticsearch实例ID，在基本信息页面获取公网地址，即为ES集群地址。
            RestClientBuilder builder = RestClient.builder(new HttpHost(this.clusterHost, this.clusterPort))
                    .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                        @Override
                        public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                            return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                        }
                    });
            // RestHighLevelClient实例通过REST low-level client builder进行构造。
            RestHighLevelClient highClient = new RestHighLevelClient(builder);
            esClientMap.put(ElasticSearchConstant.ES_CLIENT, highClient);
        }
        return esClientMap.get(ElasticSearchConstant.ES_CLIENT);
    }

    /**
     * @desc 创建索引
     */
    public void createIndex(String index) throws IOException {
        if(!existsIndex(index)) {
            CreateIndexRequest request = new CreateIndexRequest(index);
            CreateIndexResponse createIndexResponse = getEsClient().indices().create(request, COMMON_OPTIONS);
            LOGGER.info("createIndex: {}", JSON.toJSONString(createIndexResponse));
        }
    }

    /**
     * @desc 判断索引是否存在
     */
    public boolean existsIndex(String index) throws IOException {
        GetIndexRequest request = new GetIndexRequest(index);
        boolean exists = getEsClient().indices().exists(request, COMMON_OPTIONS);
        LOGGER.info("existsIndex: {}", exists);
        return exists;
    }

    /**
     * @desc 判断记录是否存在
     */
    public boolean exists(String index, String type, CusDemoInfoDocument document) throws IOException {
        GetRequest getRequest = new GetRequest(index, type, document.getId());
        getRequest.fetchSourceContext(new FetchSourceContext(false));
        getRequest.storedFields("_none_");
        boolean exists = getEsClient().exists(getRequest, COMMON_OPTIONS);
        LOGGER.info("exists: {}", exists);
        return exists;
    }

    /**
     * @desc 删除索引
     */
    public boolean deleteIndex(String index) {
        try {
            DeleteIndexRequest request = new DeleteIndexRequest(index);
            request.indicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN);
            AcknowledgedResponse deleteIndexResponse = getEsClient().indices().delete(request, RequestOptions.DEFAULT);
            return deleteIndexResponse.isAcknowledged();
        } catch (Exception e) {
            LOGGER.error("删除索引失败, index:{}", index);
            return false;
        }
    }

    /**
     * @desc 添加数据
     */
    public String addData(CusDemoInfoDocument document, String index, String type, String docId) {
        String id = null;
        try {
            //index_name为索引名称；type_name为类型名称,7.0及以上版本必须为_doc；doc_id为文档的id。
            // 同步执行，并使用自定义RequestOptions（COMMON_OPTIONS）。
            RestHighLevelClient highClient = getEsClient();
            IndexRequest request = new IndexRequest(index, type)
                    .id(docId).source(JSON.toJSONString(document), XContentType.JSON);
            request.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);// 立即刷新
            IndexResponse response = highClient.index(request, COMMON_OPTIONS);
            id = response.getId();
            //highClient.close();
            LOGGER.info("索引:{}, 数据添加, 返回码:{}, id:{}", index, response.status().getStatus(), id);
        } catch (IOException e) {
            LOGGER.error("添加数据失败, index:{}, id:{}", index, id);
        }

        return id;
    }

    /**
     * @desc 修改数据
     */
    public String updateData(CusDemoInfoDocument document, String index, String type, String docId) {
        String id = null;
        try {
            RestHighLevelClient highClient = getEsClient();
            UpdateRequest request = new UpdateRequest(index, type, docId)
                    .doc(JSON.toJSONString(document), XContentType.JSON);
            request.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
            UpdateResponse response = highClient.update(request, COMMON_OPTIONS);
            id = response.getId();
            //highClient.close();
            LOGGER.info("数据更新, 返回码:{}, id:{}", response.status().getStatus(), id);
        } catch (IOException e) {
            LOGGER.error("数据更新失败, index:{}, id:{}", index, id);
        }
        return id;
    }

    /**
     * @desc 批量插入数据
     */
    public boolean insertBatch(String index, String type, List<CusDemoInfoDocument> list) {
        BulkRequest request = new BulkRequest();
        request.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
        list.forEach(item -> request.add(new IndexRequest(index, type)
                .id(item.getId()).source(JSON.toJSONString(item), XContentType.JSON)));
        try {
            RestHighLevelClient highClient = getEsClient();
            BulkResponse bulk = highClient.bulk(request, COMMON_OPTIONS);
            bulk.status().getStatus();
            //highClient.close();
            LOGGER.info("索引:{}, 批量插入 {} 条数据成功!", index, list.size());
        } catch (IOException e) {
            LOGGER.error(e.getMessage());
            LOGGER.error("索引:{}, 批量插入数据失败", index);
            return false;
        }

        return true;
    }

    /**
     * @desc 根据id删除数据
     */
    public boolean deleteById(String index, String type, String docId) {
        DeleteRequest request = new DeleteRequest(index, type, docId);
        request.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
        try {
            DeleteResponse response = getEsClient().delete(request, COMMON_OPTIONS);
            response.status().getStatus();
            LOGGER.info("索引:{}, 根据id {} 删除数据:{}", index, docId, JSON.toJSONString(response));
        } catch (Exception e) {
            LOGGER.error("根据id删除数据失败, index:{}, id:{}", index, docId);
            return false;
        }

        return true;
    }
}